# -*- coding:utf-8 -*-
# @Author: shenyuyu
# @Time: 2023/6/30 9:46
# @File: qu_1_sparksql窗口函数.py
from pyspark.sql import SparkSession
# 导入StructType对象
from pyspark.sql.types import ArrayType, StringType, StructType, IntegerType
from pyspark.sql import functions as F

if __name__ == '__main__':
    spark = SparkSession.builder \
        .appName("create df") \
        .master("local[*]") \
        .config("spark.sql.shuffle.partitions", "2") \
        .getOrCreate()
    sc = spark.sparkContext

    rdd = sc.parallelize([
        ('张三', 'class_1', 99),
        ('王五', 'class_2', 35),
        ('王三', 'class_3', 57),
        ('王久', 'class_4', 12),
        ('王丽', 'class_5', 99),
        ('王娟', 'class_1', 90),
        ('王军', 'class_2', 91),
        ('王俊', 'class_3', 33),
        ('王君', 'class_4', 55),
        ('王珺', 'class_5', 66),
        ('郑颖', 'class_1', 11),
        ('郑辉', 'class_2', 33),
        ('张丽', 'class_3', 36),
        ('张张', 'class_4', 79),
        ('黄凯', 'class_5', 90),
        ('黄开', 'class_1', 90),
        ('黄恺', 'class_2', 90),
        ('王凯', 'class_3', 11),
        ('王凯杰', 'class_1', 11),
        ('王开杰', 'class_2', 3),
        ('王景亮', 'class_3', 99)
    ])
    schema = StructType().add("name", StringType()).add("class", StringType()). \
        add("score", IntegerType())
    df = rdd.toDF(schema)

    df.createOrReplaceTempView("stu")

    # 求平均分
    # spark.sql("select *, avg(score) over() from stu ").show()

    # spark.sql("select *, avg(score) over(partition by class) from stu ").show()

    # 排名 正常排序
    # spark.sql("select *,row_number() over(order by score ) from stu").show()
    # # 会占用
    # spark.sql("select *,rank() over(order by score ) from stu").show()
    # # 不会占用
    # spark.sql("select *,dense_rank() over(order by score ) from stu").show()
    # spark.sql("select *,dense_rank() over(partition by class order by score ) from stu").show()

    #分组
    spark.sql("select *,ntile(6) over(order by score) from stu").show()